package com.atguigu.flink.wordcount;

import com.atguigu.flink.function.MyWordCountFlatMapFunction;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/23
 *
 *  flink编程，一般调用内置的算子进行处理，大部分算子都需要传入一个 xxxFunction。
 *  xxxFunction代表这个算子运算的具体逻辑，一般都是一个函数式接口。
 *
 *      函数式接口的实现方式:
 *              ①编写外部类 实现 接口
 *              ②编写内部类 实现 接口
 *              ③编写匿名内部类 (适合运算逻辑只用一次，不复用的情况)
 *              ④使用lamda表达式
 */
public class Demo6_Function
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        DataStreamSource<String> source = env.readTextFile("data/words.txt");


        SingleOutputStreamOperator<Demo5_POJO.WordCount> ds1 = source
            .flatMap(new MyWordCountFlatMapFunction2());

        ds1
            .keyBy(new KeySelector<Demo5_POJO.WordCount, String>()
            {
                @Override
                public String getKey(Demo5_POJO.WordCount value) throws Exception {
                    return value.getWord();
                }
            })
            //写要sum的字段
            .sum("count")
            .print();

        env.execute();

    }

    public static class MyWordCountFlatMapFunction2 implements FlatMapFunction<String, Demo5_POJO.WordCount>
    {
        @Override
        public void flatMap(String line, Collector<Demo5_POJO.WordCount> out) throws Exception {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(new Demo5_POJO.WordCount(word,1));
            }
        }
    }

}
